package com.example.bigdata.kafka.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;
import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;

public class StreamSample {
    private static final String INPUT_TOPIC = "kafkaStreamIn";
    private static final String OUT_TOPIC = "kafkaStreamOut";
    public static void main(String[] args) {
        Properties props = streamInit();
        // 如何构建流结构拓扑
        final StreamsBuilder builder = new StreamsBuilder();
        wordcountStream(builder);
        final KafkaStreams streams = new KafkaStreams(builder.build(),props);
        streams.start();
    }

    /**
     *  定义一个流计算过程
     */
    static void wordcountStream(final StreamsBuilder builder){
        KStream<String, String> source = builder.stream(INPUT_TOPIC);
        final KTable<String, Long> count = source
                .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" ")))
                .groupBy((key, value) -> value)
                .count();
        count.toStream().to(OUT_TOPIC, Produced.with(Serdes.String(),Serdes.Long()));
    }
    public static Properties streamInit(){
        Properties props = new Properties();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"172.23.6.159:9092");
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,"wordCountApp");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        return props;
    }
}
